package hasthi.manager;

import static hasthi.common.constants.HasthiConstants.ELECTED_NOTICE_ACTION;
import static hasthi.common.constants.HasthiConstants.ELECTION_CANCEL_NOTICE_ACTION;
import static hasthi.common.constants.HasthiConstants.NOMINATION_REQUEST_ACTION;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.RemoteProcessorException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.constants.HasthiConstants;
import hasthi.container.SOAPUtils;
import hasthi.container.p2p.MulticastContianier;
import hasthi.manager.CoordinatorInvitaionStatus.Enum;
import hasthi.manager.CoordinatorStepDownNoticeDocument.CoordinatorStepDownNotice;
import hasthi.manager.NewCoordinatorInvitationResponseDocument.NewCoordinatorInvitationResponse;
import hasthi.manager.NominationResponseDocument.NominationResponse;
import hasthi.manager.monitering.VerificationTools;
import hasthi.tools.ManagementRingTools;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlObject;

public class ElectionController {
    protected static Log log = LogFactory.getLog(HasthiConstants.AGENT_LOGGER_NAME);
    private final ManagerContext managerContext;
    private final DistributedLogger dlogger;
    
   
    private final ExecutorService singleThreadExecutor;
    private long lastElectionTime = -1;
    @SuppressWarnings("unchecked")
    private Future electionHandle;
    
    public ElectionController(ManagerContext managerContext) {
        this.managerContext = managerContext;
        dlogger = managerContext.getBaseContext().getDlogger();
        
        singleThreadExecutor = Executors.newCachedThreadPool();
    }

    public synchronized boolean startElection(ManagerAddress oldCoordinator, long oldCoordinatorTimestamp,boolean sucessorBased){
        if(!sucessorBased){
            singleThreadExecutor.submit(new ManagerElection(oldCoordinator,oldCoordinatorTimestamp,sucessorBased));
            log.info("Submit a nomination at "+ managerContext);
            return true;
        }else{
            boolean hasElection = (lastElectionTime != -1);
            long timeTook = (System.currentTimeMillis() - lastElectionTime); 
            if(hasElection && 
                timeTook > managerContext.getBaseContext().getElectionTimeoutMs()){
                electionHandle.cancel(true);
                hasElection = false;
                log.info("Cancel a Election at "+ managerContext + " took "+ timeTook + "ms");
            }
            
            if(hasElection){
                return false;
            }else{
                lastElectionTime = System.currentTimeMillis();
                electionHandle = singleThreadExecutor.submit(new ManagerElection(oldCoordinator,oldCoordinatorTimestamp,sucessorBased));
                log.info("Submit a Election at "+ managerContext);
                return true;
            }
        }
    }
    
    
    public synchronized void electionDone(){
        lastElectionTime = -1;
    }
    
    public class ManagerElection implements Runnable{
        private ManagerAddress oldCoordinator;
        private long oldCoordinatorTimestamp;
        private boolean sucessorBased = true;
        
        
        public ManagerElection(ManagerAddress oldCoordinator, long oldCoordinatorTimestamp,
                boolean sucessorBased) {
            this.oldCoordinator = oldCoordinator;
            this.oldCoordinatorTimestamp = oldCoordinatorTimestamp;
            this.sucessorBased = sucessorBased;
        }


        public void run() {
            try {
                long electionResponseTime = managerContext.getBaseContext().getElectionResponseTimeoutMs();
                int managmentRingSize = managerContext.getManagementRingSize();
                
                int otherManagerCountApproximation = managerContext.getManagementRingSize() - 2; //removed failed coordinator and Myself 
                
                if(!sucessorBased){
                    dlogger.info("Call for nomination Started at "+ managerContext);
                    MulticastContianier p2pContainer = managerContext.getMulticastContianer();
                    XmlObject nominationRequest = ManagerClientUtils.createNominationRequest(oldCoordinator,oldCoordinatorTimestamp);
                    
                    
                    int recivedResponseCount = 0;
                    int repeatCount = 0;
                    Collection<XmlObject> responses = null;
                    while(repeatCount < 2 && 
                            ((float)recivedResponseCount)/otherManagerCountApproximation < 0.5){
                        //if not first time, we wait
                        if(repeatCount != 0){
                            try {
                                Thread.sleep(5000);
                            } catch (InterruptedException e) {
                                log.error(e);
                            }
                        }
                        responses = p2pContainer.mulitiCastAndWait(nominationRequest, 
                                NOMINATION_REQUEST_ACTION,electionResponseTime,managmentRingSize); 
                        
                        recivedResponseCount = responses.size();
                        repeatCount++;
                    }    
                    dlogger.info("Nomination responses("+recivedResponseCount +"/"+otherManagerCountApproximation+" responses)@"+ managerContext);
                    
                    
                    StringBuffer buf = new StringBuffer();
                   
                    //We start with my address, so that one is considered as well
                    int bestCandidateRank = managerContext.getMyRank();
                    ManagerAddress bestCondidateAddress = managerContext.getMyAddress();
                    buf.append(bestCandidateRank + " ");
                    
                    Map<String, Integer> coordinatorMap = new HashMap<String, Integer>();
                    for(XmlObject response:responses){
                        if(SOAPUtils.isFaultMessage(response)){
                            log.error(HasthiException.buildException(response));
                        }else{
                            NominationResponseDocument  nominationResponseDocument = (NominationResponseDocument)response;
                            NominationResponse nominationResponse = nominationResponseDocument.getNominationResponse();
                            ManagerAddress senderAddress = nominationResponse.getMyAddress();
                            
                            if(senderAddress != null){
                                buf.append(senderAddress.getRank() + " ");
                                //We find the best condidate
                                if(senderAddress.getRank() < bestCandidateRank){
                                    bestCandidateRank = senderAddress.getRank();
                                    bestCondidateAddress = senderAddress;
                                }
                                //We find existing coordinators and how many votes they had
                                if(nominationResponse.getCurrentCoordinatorAddress() != null){
                                    ManagerAddress currentCoordinator = nominationResponse.getCurrentCoordinatorAddress();
                                    String key = currentCoordinator.getSoapAddress();
                                    Integer count = coordinatorMap.get(key);
                                    if(count == null){
                                        count = new Integer(0);
                                    }
                                    count = new Integer(count + 1);
                                    coordinatorMap.put(key, count);  
                                }
                            }else{
                                log.fatal("Nomination response address must not be null "+ XmlBeanUtils.prettyPrint(nominationResponseDocument));
                            }
                        }
                    }
                    boolean coordinatorUp = false;
                    log.info("NominationSummery[best Candidate =" + (bestCondidateAddress == null?null:Utils.getID(bestCondidateAddress)) + " "+ coordinatorMap + " "+
                            "["+buf.toString()+"]]");
                    System.out.println("NominationSummery[best Candidate =" +(bestCondidateAddress == null?null:Utils.getID(bestCondidateAddress)) + " "+ coordinatorMap + " "+
                            "["+buf.toString()+"]]");
                    /**
                     * There is a another coordinator, we try to join him. If he is there, all is well. We start with the one who has
                     * most votes, and workdown the list. 
                     */
                    //
                    while(!coordinatorUp && coordinatorMap.size() > 0){
                        String bestCoordinator = null;
                        int maxVotes = -1;
                        for(String key:coordinatorMap.keySet()){
                            int votes = coordinatorMap.get(key);
                            if(votes > maxVotes){
                                maxVotes = votes;
                                bestCoordinator = key;
                            }
                        }
                        try {
                            ManagementRingTools.joinUsingHeartbeat(managerContext, bestCoordinator);
                            dlogger.info("Nomination Ending at "+ managerContext + " found Coordinator "+ bestCoordinator);
                            coordinatorUp = true;
                        } catch (Throwable e) {
                            log.warn(e);
                            coordinatorMap.remove(bestCoordinator);
                            log.info("point2");
                        }    
                    }
                    
                    if(!coordinatorUp){
                        /**
                         * No existing coordinators, we will find the best candidate and invite him
                         */
                        dlogger.info("Best Candidate "+ Utils.getID(bestCondidateAddress) + " selected, Inviting to become the coordinator");
                        inviteToBeCoordinator(bestCondidateAddress);
                    }
                }else{
                    try {
                        managerContext.setNominted(true);
                        dlogger.info("Election Started at "+managerContext);
                        ManagerAddress bestCandidateAddress = managerContext.getMyAddress();
                        /* Add this Election to the list */
                        
                        long coordinatorTimestamp = System.currentTimeMillis();
                        //to make sure timestamps are monotonically increasing
                        if(coordinatorTimestamp < managerContext.getCoordinatorTimeStamp()){
                            coordinatorTimestamp = managerContext.getCoordinatorTimeStamp() + 100;
                        }
                        //dlogger.info("Elected Notice("+bestCandidateAddress.getSoapAddress() + "/"+ coordinatorTimestamp +")");
                        XmlObject nomiatedRequest = ManagerClientUtils.createElectedNotice(bestCandidateAddress, 
                                coordinatorTimestamp);
                        MulticastContianier container = managerContext.getMulticastContianer();
                        
                        
                        int recivedResponseCount = 0;
                        /**
                         * Peer count is at least two less, myself and manager which is down 
                         */
                        int minResponseCount = (int)Math.ceil(otherManagerCountApproximation*.5);
                        int round = 0;
                        int rejectionCount = 0 ;
                        ManagerAddress alternativeAddress = null;
                        long alternativeTimestamp = -1;
                        int alternativeRank = Integer.MAX_VALUE;

                        while(round < 2 && recivedResponseCount < minResponseCount){
                            if(round != 0){
                                try {
                                    Thread.sleep(5000);
                                } catch (InterruptedException e) {
                                    log.error(e);
                                }
                            }
                            dlogger.info("Election round "+ round + "@"+ managerContext);
                            Collection<XmlObject> nominatedResponses = container.mulitiCastAndWait(nomiatedRequest, 
                                    ELECTED_NOTICE_ACTION,
                                    electionResponseTime,otherManagerCountApproximation);
                            recivedResponseCount = nominatedResponses.size();
                            dlogger.info("Elected Response("+recivedResponseCount+"/"+minResponseCount +" responses received)@"+ managerContext);
                            
                            rejectionCount = 0;
                            //TODO may be we need to use all the alternative addresses sent by peers
                            for(XmlObject response:nominatedResponses){
                                if(SOAPUtils.isFaultMessage(response)){
                                    log.error(HasthiException.buildException(response));
                                }else{
                                    ElectedNoticeResponseDocument responseDocument = (ElectedNoticeResponseDocument)response;
                                    if(!ElectedReaction.ACKED.equals(responseDocument.getElectedNoticeResponse().getReaction())){
                                        ManagerAddress tempAddress = responseDocument.getElectedNoticeResponse().getAlternativeAddress();
                                        if(tempAddress != null){
                                            if(tempAddress.getRank() < alternativeRank){
                                                alternativeAddress = tempAddress;
                                                alternativeRank = alternativeAddress.getRank();
                                                alternativeTimestamp = responseDocument.getElectedNoticeResponse().getAlternativeCoordinatorTimestamp();
                                            }
                                        }else{
                                            log.fatal("Rejects the election, but does not send a alternative");
                                        }
                                        rejectionCount++;
                                    }
                                }
                            }
                            round++;
                        }
                        
                        if(rejectionCount > 0 && alternativeRank < managerContext.getMyRank()){
                            /**
                             * There is a better option, we invite him
                             */
                            CoordinatorStepDownNoticeDocument cancelNoticeDocument = CoordinatorStepDownNoticeDocument.Factory.newInstance();
                            CoordinatorStepDownNotice coordinatorStepDownNotice = cancelNoticeDocument.addNewCoordinatorStepDownNotice();
                            coordinatorStepDownNotice.setSteppedDownCoordinator(managerContext.getMyAddress());
                            if(alternativeTimestamp > 0){
                                coordinatorStepDownNotice.setAppointedCoordinator(alternativeAddress);
                                coordinatorStepDownNotice.setAppointedCoordinatorTimestamp(alternativeTimestamp);
                            }
                            container.mulitiCastAndWait(cancelNoticeDocument, ELECTION_CANCEL_NOTICE_ACTION, 0, 0);
                            dlogger.info("Election Cancelled at "+managerContext + " favour of "+ Utils.getID(alternativeAddress));
                            if(!managerContext.hasCoordinator()){
                                inviteToBeCoordinator(alternativeAddress);
                            }
                        }else{
                            managerContext.becomeCoordinator(coordinatorTimestamp);
                            //TODO collect data about manager assignments
                            dlogger.info("Election Completed:" + managerContext + " become the coordinator");
                        }
                    } finally{
                        electionDone();
                    }
                    
                }
            } catch (Throwable e) {
               dlogger.errror("Faluire while election "+ e.getMessage(), e);
            } 
        }


        private void inviteToBeCoordinator(ManagerAddress bestCondidateAddress)
                throws RemoteProcessorException, HasthiException {
            NewCoordinatorInvitationResponseDocument responseDocument 
                = ManagementRingTools.inviteTobeCoordinator(bestCondidateAddress, managerContext,oldCoordinator,oldCoordinatorTimestamp);
            NewCoordinatorInvitationResponse coordinatorInvitationResponse = responseDocument.getNewCoordinatorInvitationResponse();
            Enum status = coordinatorInvitationResponse.getStatus();
            if(status.equals(CoordinatorInvitaionStatus.IAM_ALREADY_COORDINATOR)){
                managerContext.setCoordinatorAddress(coordinatorInvitationResponse.getCoordinatorAddress(), 
                        coordinatorInvitationResponse.getCoordinatorTimestamp());
            }else if(status.equals(CoordinatorInvitaionStatus.ELECTION_ALREADY_STARTED)){
                log.info("Elecion laredy started at "+Utils.getID(bestCondidateAddress));
                if(!managerContext.hasCoordinator()){
                    try {
                        ManagementRingTools.joinUsingHeartbeat(managerContext, bestCondidateAddress.getSoapAddress());
                    } catch (HasthiException e) {
                        log.info("Failed to Join the coordinator, joining while election is already started "+ e.getMessage());
                    }
                }
            }else if(status.equals(CoordinatorInvitaionStatus.COORDINATOR_ALREADY_EXISTS)
                    || status.equals(CoordinatorInvitaionStatus.BETTER_SUCESSOR_EXISTS)
                    || status.equals(CoordinatorInvitaionStatus.COORDINATOR_NOT_DOWN)){
                dlogger.errror("After nomination, the sucessor rejects the election "+ status,null);
            }
        }

    }
    
}
